package com.atguigu.day08;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/4/9
 * 该案例演示了算子状态的ListState
 * 需求：在map算子中每个并行度上计算数据的个数
 */
public class Flink08_OpeState_ListState {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);
        //TODO 2.从指定的网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        socketDS.process(
                new MyPro()
        ).print();


        //TODO 7.提交作业
        env.execute();
    }
}



class MyPro extends ProcessFunction<String, String> implements CheckpointedFunction{
    Integer count = 0;
    ListState<Integer> countListState;
    @Override
    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
        count ++;
        out.collect("当前并行度"+getRuntimeContext().getIndexOfThisSubtask()+"元素的个数" + count);
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        System.out.println("~~~snapshotState~~~");
        //清空状态
        countListState.clear();
        //将要备份的数据放到状态中
        countListState.add(count);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("~~~initializeState~~~");
        ListStateDescriptor<Integer> listStateDescriptor
                = new ListStateDescriptor<Integer>("countListState", Integer.class);
        countListState = context.getOperatorStateStore().getListState(listStateDescriptor);
        if(context.isRestored()){
            count = countListState.get().iterator().next();
        }
    }
}
